package workpool

import (
	"context"
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

// Task 表示一个任务
type Task struct {
	ID       int
	Priority int          // 任务优先级
	Handler  func() error // 任务处理函数
	Done     chan error   // 任务完成通知通道
}

// WorkerPool 表示一个工作池
type WorkerPool struct {
	maxWorkers     int32              // 最大worker数量
	minWorkers     int32              // 最小worker数量
	currentWorkers int32              // 当前worker数量
	taskQueue      chan *Task         // 任务队列
	stopCh         chan struct{}      // 停止信号
	workerWg       sync.WaitGroup     // worker等待组
	metrics        *Metrics           // 指标收集
	ctx            context.Context    // 上下文
	cancel         context.CancelFunc // 取消函数
}

// Metrics 用于收集指标
type Metrics struct {
	totalTasks      int64 // 总任务数
	completedTasks  int64 // 完成任务数
	failedTasks     int64 // 失败任务数
	processingTasks int32 // 正在处理的任务数
	queueLength     int32 // 队列长度
}

// NewWorkerPool 创建新的工作池
func NewWorkerPool(minWorkers, maxWorkers int, queueSize int) *WorkerPool {
	ctx, cancel := context.WithCancel(context.Background())
	wp := &WorkerPool{
		maxWorkers:     int32(maxWorkers),
		minWorkers:     int32(minWorkers),
		currentWorkers: 0,
		taskQueue:      make(chan *Task, queueSize),
		stopCh:         make(chan struct{}),
		metrics: &Metrics{
			totalTasks:      0,
			completedTasks:  0,
			failedTasks:     0,
			processingTasks: 0,
			queueLength:     0,
		},
		ctx:    ctx,
		cancel: cancel,
	}

	// 启动最小数量的worker
	for i := 0; i < minWorkers; i++ {
		wp.addWorker()
	}

	// 启动自动扩缩容
	go wp.autoScale()

	// 启动指标收集
	go wp.collectMetrics()

	return wp
}

// addWorker 添加一个worker
func (wp *WorkerPool) addWorker() {
	atomic.AddInt32(&wp.currentWorkers, 1)
	wp.workerWg.Add(1)

	go func() {
		defer wp.workerWg.Done()
		defer atomic.AddInt32(&wp.currentWorkers, -1)

		for {
			select {
			case task := <-wp.taskQueue:
				if task == nil {
					return
				}

				// 更新指标
				atomic.AddInt32(&wp.metrics.processingTasks, 1)

				// 执行任务
				err := task.Handler()

				// 更新指标
				atomic.AddInt32(&wp.metrics.processingTasks, -1)
				if err != nil {
					atomic.AddInt64(&wp.metrics.failedTasks, 1)
				} else {
					atomic.AddInt64(&wp.metrics.completedTasks, 1)
				}

				// 通知任务完成
				if task.Done != nil {
					task.Done <- err
					close(task.Done)
				}

			case <-wp.ctx.Done():
				return
			}
		}
	}()
}

// Submit 提交任务
func (wp *WorkerPool) Submit(task *Task) error {
	select {
	case <-wp.ctx.Done():
		return fmt.Errorf("worker pool is stopped")
	case wp.taskQueue <- task:
		atomic.AddInt64(&wp.metrics.totalTasks, 1)
		atomic.AddInt32(&wp.metrics.queueLength, 1)
		return nil
	}
}

// Stop 停止工作池
func (wp *WorkerPool) Stop() {
	wp.cancel()
	close(wp.taskQueue)
	wp.workerWg.Wait()
}

// autoScale 自动扩缩容
func (wp *WorkerPool) autoScale() {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			queueLen := atomic.LoadInt32(&wp.metrics.queueLength)
			currentWorkers := atomic.LoadInt32(&wp.currentWorkers)
			processingTasks := atomic.LoadInt32(&wp.metrics.processingTasks)

			// 根据队列长度和处理中的任务数决定是否需要扩容或缩容
			if queueLen > 0 && currentWorkers < wp.maxWorkers &&
				float64(processingTasks)/float64(currentWorkers) > 0.8 {
				// 扩容
				wp.addWorker()
			} else if queueLen == 0 && currentWorkers > wp.minWorkers &&
				float64(processingTasks)/float64(currentWorkers) < 0.2 {
				// 缩容
				atomic.AddInt32(&wp.currentWorkers, -1)
			}

		case <-wp.ctx.Done():
			return
		}
	}
}

// collectMetrics 收集指标
func (wp *WorkerPool) collectMetrics() {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			log.Printf("Worker Pool Metrics:\n"+
				"Total Tasks: %d\n"+
				"Completed Tasks: %d\n"+
				"Failed Tasks: %d\n"+
				"Processing Tasks: %d\n"+
				"Queue Length: %d\n"+
				"Current Workers: %d\n",
				atomic.LoadInt64(&wp.metrics.totalTasks),
				atomic.LoadInt64(&wp.metrics.completedTasks),
				atomic.LoadInt64(&wp.metrics.failedTasks),
				atomic.LoadInt32(&wp.metrics.processingTasks),
				atomic.LoadInt32(&wp.metrics.queueLength),
				atomic.LoadInt32(&wp.currentWorkers))

		case <-wp.ctx.Done():
			return
		}
	}
}

func Test() {
	// 创建工作池
	pool := NewWorkerPool(2, 10, 100)

	// 模拟提交任务
	for i := 0; i < 20; i++ {
		taskID := i
		task := &Task{
			ID: taskID,
			Handler: func() error {
				// 模拟任务处理
				time.Sleep(time.Second)
				log.Printf("Task %d completed\n", taskID)
				return nil
			},
			Done: make(chan error, 1),
		}

		if err := pool.Submit(task); err != nil {
			log.Printf("Failed to submit task %d: %v\n", taskID, err)
			continue
		}

		// 异步等待任务完成
		go func() {
			if err := <-task.Done; err != nil {
				log.Printf("Task %d failed: %v\n", taskID, err)
			}
		}()
	}

	// 运行一段时间后停止
	time.Sleep(10 * time.Second)
	pool.Stop()
}

/*

3. 重要优化建议
	任务批处理
		合并小任务减少开销
		实现批量提交接口
		优化内存分配

	负载均衡
		实现工作窃取算法
		动态调整任务分配
		避免饥饿问题

	资源管理
		实现优雅关闭
		处理panic情况
		释放资源

	监控告警
		设置健康检查
		实现自动恢复
		记录详细日志

	六、总结
	工作池的实现需要考虑以下关键点：

		基础架构
			合理的接口设计
			良好的扩展性
			完善的错误处理

		性能优化
			减少锁竞争
			优化内存使用
			提高并发效率

		可靠性
			处理边界情况
			实现容错机制
			保证数据一致性

		可维护性
			清晰的代码结构
			完善的文档
			便于测试和调试
*/
